package com.cnebula.dataprocess.hive.job;

import java.util.ArrayList;
import java.util.List;

/**
 * hive 去重SQL拼接器
 * 
 * 样例sql：
 * 
 * insert overwrite table goods select id,name,pricie,type from (select
 * id,name,pricie,type, row_number() over(distribute by id sort by id) as rn
 * from goods) t where t.rn=1;
 */
public class HiveDataPassSQLBuilder {

	StringBuilder str = new StringBuilder("insert overwrite table ");

	int columnNumber = 0;
	String tableName;
	String passColumn;
	List<String> columns;

	public HiveDataPassSQLBuilder(String tableName, String passColumn, List<String> columns) {
		appendWithSpace(tableName);
		this.tableName = tableName;
		this.passColumn = passColumn;
		this.columns = columns;
	}

	private HiveDataPassSQLBuilder appendColumn(String columnName) {
		if (columnNumber != 0) {
			appendWithSpace(",");
		}
		appendWithSpace(columnName);
		columnNumber++;
		return this;
	}

	private String rowNumberFunc(String passColumn) {
		StringBuilder s = new StringBuilder("row_number()").append(" ");
		s.append("over( distribute by ");
		s.append(passColumn);
		s.append(" sort by ");
		s.append(passColumn).append("");
		s.append(") as rn");

		return s.toString();
	}

	private void appendWithSpace(String s) {
		str.append(s).append(" ");
	}

	public String getResult() {
		appendWithSpace("select");
		columns.forEach(col -> {
			appendColumn(col);
		});
		appendWithSpace("from");
		appendWithSpace("(");
		appendWithSpace("select");
		columnNumber = 0;
		columns.forEach(col -> {
			appendColumn(col);
		});
		appendColumn(rowNumberFunc(passColumn));

		appendWithSpace("from");
		appendWithSpace(tableName);
		appendWithSpace(") t where t.rn=1");

		return str.toString();
	}

	public static void main(String[] args) {
		String tableName = "goods";
		String passColumn = "id";
		List<String> columns = new ArrayList<String>();
		columns.add("id");
		columns.add("name");
		columns.add("type");
		columns.add("pricie");

		HiveDataPassSQLBuilder str = new HiveDataPassSQLBuilder(tableName, passColumn, columns);
		System.out.println(str.getResult());
	}
}